use std::any::Any;
use std::sync::Arc;

use async_trait::async_trait;
use datafusion::arrow::datatypes::SchemaRef;
use datafusion::datasource::{TableProvider, TableType};
use datafusion::execution::context::SessionState;
use datafusion::logical_expr::Expr;
use datafusion::physical_plan::ExecutionPlan;
use log::debug;

use common::LResult;
use engine::TableScanOption;

use crate::mem_table::MemTableRef;
use crate::mem_table_scan::MemTableScan;

pub struct MemTableProvider {
    mem_table: MemTableRef,
    schema: SchemaRef,
}

impl MemTableProvider {
    pub fn create(mem_table: MemTableRef) -> LResult<Self> {
        let schema = Arc::new(mem_table.read().schema.as_schema());
        Ok(Self {
            mem_table,
            schema,
        })
    }
}

#[async_trait]
impl TableProvider for MemTableProvider {
    fn as_any(&self) -> &dyn Any {
        self
    }

    fn schema(&self) -> SchemaRef {
        self.schema.clone()
    }

    fn table_type(&self) -> TableType {
        TableType::Base
    }
    async fn scan(&self, _state: &SessionState, projection: Option<&Vec<usize>>, filters: &[Expr], limit: Option<usize>) -> datafusion::common::Result<Arc<dyn ExecutionPlan>> {
        #[cfg(debug_assertions)]
        {
            debug!(target: "engine:mem", "scan {}, projection: {:?}, filter: {:?}, limit: {:?}", self.mem_table.read().measurement(), projection, filters, limit);
        }
        let projection = projection.map(|v|v.clone());
        Ok(Arc::new(MemTableScan::create(
            self.mem_table.clone(),
            self.schema.clone(),
            Arc::new(TableScanOption {
                projection,
                limit
            })
        )))
    }
}